<?php

namespace emqx;

use Bluerhinos\phpMQTT;
use job\ThinkQueueJob;

class EmqxMqttClient extends Emqx
{
    //MQTT客户端ID
    public string $mqtt_client_id;
    //订阅主题
    public array $mqtt_topics;

    /**
     * 初始化
     * @return void
     */
    public function __construct()
    {
        parent::__construct();
        //mqtt客户端ID
        $this->mqtt_client_id = 'mqtt_iot_' . uniqid();
        //订阅主题
        $this->mqtt_topics = [
            //接收设备消息主题
            $this->emqx_mqtt_device_topic => array('qos' => 0, 'function' => [$this, 'procClientMsg']),
            //EMQX系统主题
            //上线和离线
            '$SYS/brokers/+/clients/+/connected' => array('qos' => 0, 'function' => [$this, 'procSysMsgConnected']),
            '$SYS/brokers/+/clients/+/disconnected' => array('qos' => 0, 'function' => [$this, 'procSysMsgDisconnected']),
        ];
    }

    /**
     * MQTT发送消息
     * @param string $topic
     * @param string $content
     * @param int $qos
     * @return bool
     */
    public function publishMqtt(string $topic, string $content, int $qos = 0): bool
    {
        $mqtt = new phpMQTT($this->emqx_mqtt_host, $this->emqx_mqtt_port, $this->mqtt_client_id);
        if ($mqtt->connect(true, NULL, $this->emqx_mqtt_php_username, $this->emqx_mqtt_php_password)) {
            //如果为数组则转为json字符串
            $mqtt->publish($topic, $content, $qos, false);
            $mqtt->close();
        } else {
            return false;
        }
        return true;
    }

    /**
     * MQTT订阅主题
     * @return void
     */
    public function subscribeMqtt(int $qos = 0)
    {
        $mqtt = new phpMQTT($this->emqx_mqtt_host, $this->emqx_mqtt_port, $this->mqtt_client_id);
        if (!$mqtt->connect(true, NULL, $this->emqx_mqtt_php_username, $this->emqx_mqtt_php_password)) {
            exit(1);
        }
        $mqtt->debug = false;
        $mqtt->subscribe($this->mqtt_topics, $qos);
        while ($mqtt->proc()) {

        }
        $mqtt->close();
    }

    /**
     * 处理订阅消息(推送至消息队列)
     * @param string $topic
     * @param mixed $msg
     * @return void
     */
    public function procClientMsg(string $topic, mixed $msg)
    {
        ThinkQueueJob::sendMsg('EmqxMqttClient', $topic, $msg);
    }

    /**
     * 客户端上线(推送至消息队列)
     * @param string $topic
     * @param mixed $msg
     * @return void
     */
    public function procSysMsgConnected(string $topic, mixed $msg)
    {
        ThinkQueueJob::sendMsg('EmqxSysMsgConnected', $topic, $msg);
    }

    /**
     * 客户端离线(推送至消息队列)
     * @param string $topic
     * @param mixed $msg
     * @return void
     */
    public function procSysMsgDisconnected(string $topic, mixed $msg)
    {
        ThinkQueueJob::sendMsg('EmqxSysMsgDisconnected', $topic, $msg);
    }

}